package com.haoziqi.chapter_02;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * description
 * created by A on 2021/3/4
 */
public class MySource {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new MyFunction()).print();
        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
    public static class MyFunction implements SourceFunction<WaterSensor>{
        private static volatile boolean runningFlag=true;
        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {
            while(runningFlag){
                Random random = new Random();
                sourceContext.collect(
                        new WaterSensor("sensor_"+random.nextInt(3),
                                System.currentTimeMillis(),
                                random.nextInt(10))
                );
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            runningFlag=false;
        }
    }
}
